package com.soda.utils;

import com.coreos.jetcd.*;
import com.coreos.jetcd.data.ByteSequence;
import com.coreos.jetcd.kv.GetResponse;
import com.coreos.jetcd.kv.TxnResponse;
import com.coreos.jetcd.lease.LeaseGrantResponse;
import com.coreos.jetcd.op.Cmp;
import com.coreos.jetcd.op.CmpTarget;
import com.coreos.jetcd.op.Op;
import com.coreos.jetcd.options.GetOption;
import com.coreos.jetcd.options.PutOption;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang3.StringUtils;
import org.springframework.beans.factory.InitializingBean;

import java.net.InetAddress;
import java.net.UnknownHostException;
import java.util.concurrent.*;

/**
 * Created by life on 2018/3/22.
 *
 */
@Slf4j
public class ServerRegister implements InitializingBean {
    //单作业线程池
    private static final ExecutorService EXECUTOR_HOLDING_SERVICE = Executors.newSingleThreadExecutor();
    private static final ScheduledExecutorService SCHEDULED_MONITOR_SERVICE = Executors.newSingleThreadScheduledExecutor();
    //index目录路径
    private static final String INDEX_PATH_TAIL = "/indexs/index_";
    //app名称
    private String appName;
    //相应密码
    private String password;
    //
    private String INDEX_PATH;
    //本机主地址
    private String HOST_ADDRESS = null;
    //服务器的地址
    private String[] etcdUrlsArr = null;
    //单行形式
    private String etcdUrls = null;
    //
    public String getEtcdUrls() {
        return etcdUrls;
    }
    //
    public void setEtcdUrls(String etcdUrls) {
        this.etcdUrls = etcdUrls;
    }
    //
    public String getAppName() {
        return appName;
    }
    //
    public void setAppName(String appName) {
        this.appName = appName;
    }
    //
    public String getPassword() {
        return password;
    }
    //
    public void setPassword(String password) {
        this.password = password;
    }
    //
    private IndexHolder indexHolder = null;
    //获取etcd客户端
    public Client getClient() {
        return indexHolder.client;
    }
    //获取服务注册的角标index
    public Integer getIndex() {
        return indexHolder.index;
    }
    //获取所有url
    public String[] getEtcdUrlsArr() {
        return etcdUrlsArr;
    }
    /**
     *
     * @throws UnknownHostException
     */
    public ServerRegister() {
    }

    @Override
    public void afterPropertiesSet() throws Exception {
        //
        this.HOST_ADDRESS = InetAddress.getLocalHost().getHostAddress();
        this.INDEX_PATH = "/" + appName + INDEX_PATH_TAIL;
        this.etcdUrlsArr = etcdUrls.split("\\s*,\\s*");
        instantiate(false);
    }

    /**
     *
     */
    public void instantiate(boolean isFirst) throws ExecutionException, InterruptedException {
        try {
            indexHolder = new IndexHolder();
            indexHolder.exec();
        } catch (Exception ex) {
            log.error(String.format("获取index失败！进行清空操作！"), ex);
            if(null != indexHolder && null != indexHolder.client) {
                indexHolder.client.close();
            }
            throw new RuntimeException(String.format("获取index失败！信息：%1s", ex.getMessage()), ex);
        }
        //
        EXECUTOR_HOLDING_SERVICE.submit(new Runnable() {
            @Override
            public void run() {
                try {
                    indexHolder.holding();
                } catch (Exception e) {
                    log.error(String.format("Holding失败！%1s", e.getMessage()), e);
                    //尝试进入重试流程
                    if(null != indexHolder.client) {
                        indexHolder.client.close();
                        indexHolder.isAlive = false;
                    }
                }
            }
        });
        //
        if(isFirst) {
            SCHEDULED_MONITOR_SERVICE.scheduleAtFixedRate(new Runnable() {
                @Override
                public void run() {
                    if(null == indexHolder || !indexHolder.isAlive) {
                        log.info("client holding alive 丢失！进入重试流程。");
                        try {
                            instantiate(false);
                        } catch (Exception e) {
                        }
                    } else {
                        log.debug("client holding alive 正常");
                    }
                }
            }, 60, 60, TimeUnit.SECONDS);
        }
    }

    /**
     *
     * holding的关键内部类
     */
    private class IndexHolder {
        //
        public boolean isAlive;
        //
        public Long leaseId;
        //客户端
        public Client client;
        //注册服务的下标
        public Integer index;
        //
        public Lease.KeepAliveListener keepAliveListener;
        //
        public IndexHolder exec() {
            for(int i = 1; i <= 5; i++) {
                log.info(String.format("获取客户端的第%1s次尝试！", i));
                try {
                    ClientBuilder builder = Client.builder().endpoints(etcdUrlsArr);
                    if(StringUtils.isNotEmpty(password)) {
                        builder = builder.user(ByteSequence.fromString(appName)).password(ByteSequence.fromString(password));
                    }
                    client = builder.build();
                    genIndex();
                    //
                    return this;
                } catch (Exception ex) {
                    log.error(ex.getMessage(), ex);
                }
            }
            throw new RuntimeException("获取client失败！");
        }

        /**
         * 获取index
         */
        public void genIndex() throws InterruptedException, UnknownHostException, ExecutionException {
            TxnResponse txnResponse = null;
            int res = 0;
            int retryTimes = 0;
            KV kvClient = client.getKVClient();
            Lease leaseClient = client.getLeaseClient();
            //
            ByteSequence value = ByteSequence.fromString(InetAddress.getLocalHost().getHostAddress());
            do {
                Thread.sleep(500);
                //
                if(16 == res) {
                    res = 0;
                    if(3 == retryTimes++) {
                        throw new RuntimeException("重试已经超过3次，获取失败！");
                    }
                }
                ByteSequence key = ByteSequence.fromString(INDEX_PATH + res);
                Txn txn = kvClient.txn();
                CompletableFuture<LeaseGrantResponse> grant = leaseClient.grant(10);
                leaseId = grant.get().getID();
                txnResponse = txn.If(new Cmp(key, Cmp.Op.EQUAL, CmpTarget.createRevision(0)))
                        .Then(Op.put(key, value, PutOption.newBuilder().withLeaseId(leaseId).build()))
                        .Else(Op.get(key, GetOption.DEFAULT))
                        .commit().get();
                res++;
            } while(null == txnResponse || !txnResponse.isSucceeded());
            keepAliveListener = leaseClient.keepAlive(leaseId);
            index = --res;
            isAlive = true;
            log.info(String.format("争抢到的结果：%1s", res));
        }

        /**
         *
         * @throws InterruptedException
         */
        public void holding() throws InterruptedException, ExecutionException, UnknownHostException {
            while(true) {
                Thread.sleep(5000);
                log.debug(String.valueOf(keepAliveListener.listen().getTTL()));
                indexHolder.checkReplacementNecessity();
            }
        }

        /**
         * 检查位置是否有必要进行替换
         */
        public void checkReplacementNecessity() throws ExecutionException, InterruptedException, UnknownHostException {
            //
            KV kvClient = client.getKVClient();
            GetResponse my = kvClient.get(ByteSequence.fromString(INDEX_PATH), GetOption.newBuilder().withPrefix(ByteSequence.fromString(INDEX_PATH)).build()).get();
            if(index >= my.getCount()) {//如果自己的位置大于相应的位置进行位置替换操作
                log.info("前置位置空缺，重新争抢！");
                keepAliveListener.close();
                Thread.sleep(20);
                genIndex();
            }
        }
    }
}
